//
//  Agora Streaming Kit
//
//  Created by ShengQiang Liu in 2020-02.
//  Copyright (c) 2020 Agora IO. All rights reserved.
//

#include "rtmp_connection_impl.h"

#include <algorithm>
#include <memory>

#include "AgoraBase.h"
#include "api/video/i420_buffer.h"
#include "api/video/i422_buffer.h"
#include "facilities/tools/api_logger.h"
#include "rtmp_local_user_impl.h"
#include "rtmp_sender_impl.h"
#include "utils/log/log.h"

namespace agora {
namespace rtc {

const char MODULE_NAME[] = "[RtmpConnection]";

RtmpConnectionImpl::RtmpConnectionImpl()
    : connection_state_(STATE_DISCONNECTED),
      last_audio_pts_ms_(0),
      is_ready_(false),
      connection_observers_(utils::RtcSyncCallback<IRtmpConnectionObserver>::Create()),
      enable_write_flv_file_(false) {}

RtmpConnectionImpl::~RtmpConnectionImpl() { deinitialize(); }

int RtmpConnectionImpl::initialize(const RtmpConnectionConfiguration& config) {
  auto rtmp_sender = std::shared_ptr<IAgoraRtmpSender>(new RtmpSenderImpl(), [](auto it) {
    // do nothing since it's deleted by Release() method.
  });
  RtmpConfig rtmp_config;
  int rotated_width = config.videoConfig.width;
  int rotated_height = config.videoConfig.height;
  if ((config.videoConfig.orientationMode == ORIENTATION_MODE_FIXED_PORTRAIT &&
       rotated_width > rotated_height) ||
      (config.videoConfig.orientationMode == ORIENTATION_MODE_FIXED_LANDSCAPE &&
       rotated_width < rotated_height)) {
    std::swap(rotated_width, rotated_height);
  }
  if (!config.audioOnly) {
    rtmp_config.video_width = rotated_width;
    rtmp_config.video_height = rotated_height;
    rtmp_config.video_bitrate = config.videoConfig.bitrate;
    rtmp_config.framerate = config.videoConfig.framerate;
  }

  rtmp_config.audio_sample_rate_hz = config.audioConfig.sampleRateHz;
  rtmp_config.audio_bytes_per_sample = config.audioConfig.bytesPerSample;
  rtmp_config.audio_numbder_of_channels = config.audioConfig.numberOfChannels;
  rtmp_config.gop_ms = config.videoConfig.gopInMs;

  int ret = rtmp_sender->Initialize(rtmp_config);
  if (ret != 0) {
    log(commons::LOG_ERROR, "%s failed to initialize rtmp sender: %d", MODULE_NAME, ret);
    return ret;
  }

  last_audio_pts_ms_ = 0;

  enable_write_flv_file_ = config.enableWriteFlvFile;
  rtmp_sender_ = std::move(rtmp_sender);
  rtmp_sender_->RegisterRtmpSenderObserver(this);

  RtmpLocalUserImpl::Config local_user_config;
  local_user_config.rtmp_connnection = this;
  local_user_config.video_config = config.videoConfig;
  local_user_config.audio_config = config.audioConfig;

  rtmp_local_user_ = std::make_unique<RtmpLocalUserImpl>(local_user_config);

  is_ready_ = true;
  return 0;
}

int RtmpConnectionImpl::deinitialize() {
  is_ready_ = false;
  return utils::major_worker()->sync_call(LOCATION_HERE, [this] {
    log(commons::LOG_INFO, "%s deinitialize", MODULE_NAME);
    rtmp_local_user_.reset();
    if (rtmp_sender_) {
      rtmp_sender_->UnregisterRtmpSenderObserver(this);
      rtmp_sender_->Stop();
      rtmp_sender_->Release();
      rtmp_sender_.reset();
    }
    log(commons::LOG_INFO, "%s deinitialize done", MODULE_NAME);
    return 0;
  });
}

int RtmpConnectionImpl::connect(const char* url) {
  return utils::major_worker()->sync_call(LOCATION_HERE, [this, url]() {
    last_audio_pts_ms_ = 0;
    if (enable_write_flv_file_) {
      // need to start recording flv before start() so as to record the metadata.
      std::string file_path = commons::join_path(commons::get_log_path(), "flv_dump.flv");
      rtmp_sender_->WriteFlvFile(file_path.c_str(), true, true);
      // TODO(Haonong Yu): 2020/4/27 close flv if start() failed, or even better to move the flv
      //  open/close logic to rtmp sender.
    }

    rtmp_sender_->Start(url);
    return 0;
  });
}

int RtmpConnectionImpl::disconnect() {
  return utils::major_worker()->sync_call(LOCATION_HERE, [this]() {
    rtmp_sender_->Stop();

    if (enable_write_flv_file_) {
      rtmp_sender_->WriteFlvClose();
    }
#ifdef FEATURE_VIDEO
    webrtc::release_i420_buffer_cache();
    webrtc::release_i422_buffer_cache();
#endif
    return 0;
  });
}

RtmpConnectionInfo RtmpConnectionImpl::getConnectionInfo() {
  RtmpConnectionInfo info;
  info.state = connection_state_;
  return info;
}

IRtmpLocalUser* RtmpConnectionImpl::getRtmpLocalUser() { return rtmp_local_user_.get(); }

int RtmpConnectionImpl::registerObserver(IRtmpConnectionObserver* observer,
                                         void (*safeDeleter)(IRtmpConnectionObserver*)) {
  API_LOGGER_MEMBER("observer:%p", observer);
  if (!observer) return -ERR_INVALID_ARGUMENT;
  connection_observers_->Register(observer);
  return 0;
}

int RtmpConnectionImpl::unregisterObserver(IRtmpConnectionObserver* observer) {
  API_LOGGER_MEMBER("observer:%p", observer);
  if (!observer) return -ERR_INVALID_ARGUMENT;
  connection_observers_->Unregister(observer);
  return 0;
}

int RtmpConnectionImpl::SendAudioFrame(const uint8_t* data, size_t size, int64_t pts_ms) {
  pts_ms = static_cast<int64_t>(commons::tick_ms());

  if (last_audio_pts_ms_ >= pts_ms) {
    ++last_audio_pts_ms_;
  } else {
    last_audio_pts_ms_ = pts_ms;
  }

  return rtmp_sender_->SendAudioFrame(data, size, last_audio_pts_ms_);
}

int RtmpConnectionImpl::SendVideoFrame(const uint8_t* data, size_t size, int64_t pts_ms,
                                       bool is_key) {
  pts_ms = static_cast<int64_t>(commons::tick_ms());
  return rtmp_sender_->SendVideoFrame(data, size, pts_ms, is_key);
}

void RtmpConnectionImpl::OnRtmpSocketStateChanged(RtmpSocketState status) {
  if (status == kSocketStateUnknown) {
    /* code */
  } else if (status == kSocketStateStartConnected) {
    log(commons::LOG_INFO, "%s %s: rtmp connected", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_CONNECTED;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) { callback->onConnected(info); });
  } else if (status == kSocketStateSending) {
    /* code */
  } else if (status == kSocketStateStop) {
    log(commons::LOG_INFO, "%s %s: rtmp disconnected", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_DISCONNECTED;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) { callback->onDisconnected(info); });
  } else if (status == kSocketStateStartError || status == kSocketStateReconnectTimeout) {
    log(commons::LOG_ERROR, "%s %s: rtmp connect failed", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_FAILED;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) {
      callback->onConnectionFailure(info, RTMP_CONNECTION_ERR_FAILED);
    });
  } else if (status == kSocketStateReconnecting) {
    log(commons::LOG_INFO, "%s %s: rtmp reconnecting", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_RECONNECTING;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) { callback->onReconnecting(info); });
  } else if (status == kSocketStateReconnected) {
    log(commons::LOG_INFO, "%s %s: rtmp reconnected", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_RECONNECTED;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) { callback->onReconnected(info); });
  } else if (status == kSocketStateBadName) {
    log(commons::LOG_ERROR, "%s %s: rtmp connect failed", MODULE_NAME, __FUNCTION__);
    connection_state_ = STATE_FAILED;
    RtmpConnectionInfo info;
    info.state = connection_state_;
    connection_observers_->Call([this, info](auto callback) {
      callback->onConnectionFailure(info, RTMP_CONNECTION_ERR_BAD_NAME);
    });
  }
}

void RtmpConnectionImpl::OnRtmpSenderStatisticsInfo(RtmpSenderStatisticsInfo statistics_info) {
  connection_observers_->Call([this, statistics_info](auto callback) {
    callback->onTransferStatistics(
        statistics_info.video_width, statistics_info.video_height, statistics_info.video_bitrate,
        statistics_info.audio_bitrate, statistics_info.video_frame_rate,
        statistics_info.push_video_frame_cnt, statistics_info.pop_video_frame_cnt);
  });
}

void RtmpConnectionImpl::OnRtmpStreamingBufferState(RtmpStreamingBufferState status) {
  if (!is_ready_) {
    return;
  }
  agora_refptr<rtc::RtmpConnectionImpl> self = this;
  utils::major_worker()->async_call(LOCATION_HERE, [self, status]() {
    if (!self->rtmp_local_user_) {
      return -ERR_INVALID_STATE;
    }
    if (status == kStreamingBufferIncreasing) {
      self->rtmp_local_user_->adjustVideoBitrate(
          IRtmpLocalUser::VideoBitrateAdjustType::Decreasing);
    } else if (status == kStreamingBufferDeclining) {
      self->rtmp_local_user_->adjustVideoBitrate(
          IRtmpLocalUser::VideoBitrateAdjustType::Increasing);
    }
    return -ERR_OK;
  });
}

}  // namespace rtc
}  // namespace agora
